import io.vertx.core.AbstractVerticle
import mu.KotlinLogging
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently
import org.apache.rocketmq.common.consumer.ConsumeFromWhere
import java.text.SimpleDateFormat
import java.util.*

class ConsumerVerticle : AbstractVerticle() {
  private val logger = KotlinLogging.logger {}
  override fun start() {
    val consumer = DefaultMQPushConsumer("Msg1")
    consumer.namesrvAddr = "192.168.201.129:9876"
    consumer.consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
    consumer.subscribe("topic", "tags")

    consumer.registerMessageListener(MessageListenerConcurrently { msgs, context ->
      for (msg in msgs) {
        val message = String(msg.body)
        //Log.d("收到消息"+message)
        val currentTime = SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Date())
        logger.debug(currentTime.toString()+"收到消息"+message)
        // 处理消息
        vertx.eventBus().send("rocketmq_message_received", message)
      }
      ConsumeConcurrentlyStatus.CONSUME_SUCCESS
    })

    consumer.start()
  }
}
